package csl.mq.rabbitmq.receive;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.io.IOException;


/**
 * @Description: TODO
 */
//@Component
@Slf4j
public class MessageReceive {

//    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 布隆过滤器（去重）-消费成功就放进去
     */
//    @Autowired
    private BitMapBloomFilter bitMapBloomFilter;

    private static final String MESSAGE_PREFIX = "message";

    @RabbitListener(
            bindings = {
                    @QueueBinding(
                            value=@Queue,
                            key = {"error"},
                            exchange = @Exchange(
                                    value = "directs",
                                    type = ExchangeTypes.DIRECT
                            )
                    )
            }
    )
    public void receiveMessage1(String content, Message message, Channel channel){
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 消息唯一ID
        String messageId = message.getMessageProperties().getMessageId();

        if (bitMapBloomFilter.contains(messageId)){
            log.error("消息已被消费，不能重复消费！");
            try{
                channel.basicAck(deliveryTag,false);
            }catch (IOException e){
                e.printStackTrace();
            }
            return;
        }
        try{
            if ("123456".equals(content)){
                /*合格消息*/
                //签收消息-参数1：消息投送ID,参数2：是否为批量签收
                channel.basicAck(deliveryTag,false);
                log.info("消费者签收的消息---->{}:{}",messageId,content);
                /*放进布隆过滤器去重-避免重复消费*/
                bitMapBloomFilter.add(messageId);
            }else{
                /*不合格消息*/
                /*当不签收次数达到三次就不重新放入队列中*/
                String count = redisTemplate.opsForValue().get(MESSAGE_PREFIX+messageId);
                if (count!=null && Long.parseLong(count)>=3){
                    channel.basicNack(deliveryTag,false,false);
                    log.warn("消费者不签收达到三次的消息---->{}:{}",messageId,content);
                    redisTemplate.delete(MESSAGE_PREFIX+messageId);
                }
                //不签收消息-参数1：消息投送ID,参数2：是否为批量不签收，参数3：是否重新放入队列中重新消费
                channel.basicNack(deliveryTag,false,true);
                redisTemplate.opsForValue().increment(MESSAGE_PREFIX + messageId);
                log.warn("消费者不签收的消息---->{}:{}",messageId,content);
            }
        }catch (IOException e) {
            e.printStackTrace();
            log.info("消息签收异常---->{}:{}",messageId,content);
        }
    }
}
